package top.birdhk.TestAPI.state;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import top.birdhk.TestAPI.beans.SensorReading;

/**
 * 状态后端
 * 检查点设置
 */
public class StateTest_StateBackend {


    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        env.setParallelism(1);

        // 1.设置状态后端
        env.setStateBackend(new MemoryStateBackend());
        //文件系统存储,hdfs
        //env.setStateBackend(new FsStateBackend(""));
        //rocksdb存储
        //env.setStateBackend(new RocksDBStateBackend());

        // 2.检查点配置
        // 周期性保存
        env.enableCheckpointing(300);

        // 高级选项,设置检查点级别
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 设置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(60000L);
        // 设置同时进行的checkpoint保存,存在情况，一个一个来，没处理完的情况。
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        // 两次checkpoint保存的处理最小间隔，要留出时间来处理数据
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(100L);


        // 3.重启策略配置
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,10000L));  // 固定延迟重启
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.minutes(10),Time.minutes(1))); // 失败率重启


        DataStreamSource<String> dataStream = env.readTextFile("E:\\flink\\wordcount\\src\\main\\resources\\sensor.txt");

//        DataStreamSource<String> dataStream = env.socketTextStream("localhost", 7777);


        SingleOutputStreamOperator<SensorReading> inputStream = dataStream.map(line -> {
            String[] split = line.split(",");
            return new SensorReading(split[0], new Long(split[1]), new Double(split[2]));
        });


        // 定义一个有状态的map操作，统计当前sensor数据的个数
        SingleOutputStreamOperator<Integer> mapStream = inputStream
                .keyBy("id")
                .map(new StateTest_KeyedState.MyKeyCountMapper());


        mapStream.print("count:");

        env.execute();



    }


}
